package com.pie4cloud.pie.bus.api.consumer;

import com.alibaba.fastjson.JSON;
import com.pie4cloud.pie.bus.api.dto.MessageDto;
import com.pie4cloud.pie.bus.api.feign.RemoteKafkaSendService;
import com.pie4cloud.pie.common.core.constant.SecurityConstants;
import com.pie4cloud.pie.common.core.util.R;
import lombok.RequiredArgsConstructor;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicLong;

//Slf4j
@Service
@RequiredArgsConstructor
public class MetricCalcService {
	@Value("${spring.application.name}")
	private String applicationName;

	public BlockingQueue<List<MessageDto>> taskInfoQueue = new LinkedBlockingDeque<>();

	ExecutorService executorService = Executors.newFixedThreadPool(2);

	public boolean flag = true;

	public int retryNum=3;

	public AtomicLong receive = new AtomicLong(0L);

	public AtomicLong exit = new AtomicLong(0L);

	public AtomicLong success = new AtomicLong(0L);

	public AtomicLong fileWriter = new AtomicLong(0L);

	private String fileDirector="";

	private final RemoteKafkaSendService remoteKafkaSendService;

	@PostConstruct
	public void init() {
		String property = System.getProperty("user.dir");
		fileDirector=property+"/dump/"+applicationName;
		File file = new File(fileDirector);
		if (!file.exists() || !file.isDirectory()) {
			FileUtils.deleteQuietly(file);
			boolean bool1 = (new File(this.fileDirector)).mkdirs();
		}
		for (int i = 0; i < 2; i++) {
			if (i == 0) {
				this.executorService.submit((Runnable)new TaskReaderRunner());
			} else {
				this.executorService.submit((Runnable)new TaskWriterRunner(i));
			}
		}
	}

	@PreDestroy
	public void destory() {
		this.executorService.shutdown();
	}

	public class TaskReaderRunner implements Runnable{
        public TaskReaderRunner(){

		}
		@Override
		public void run() {
			while (true) {
				try {
					if (flag) {
						File file = new File(fileDirector);
						File[] fileList = file.listFiles();
						long time = new Date().getTime()-60*1000*2;
						File f = null;
						if (fileList == null)
							continue;
						for (int i = 0; i < fileList.length; i++) {
							File files = fileList[i];
							if (time > files.lastModified()) {
								f = files;
								break;
							}
						}
						if (f == null) {
							Thread.sleep(3000L);
							continue;
						}
						RandomAccessFile randomAccessFile = new RandomAccessFile(f, "rw");
						FileChannel fileChannel = randomAccessFile.getChannel();
						FileLock fileLock = null;
						while (true) {
							fileLock = fileChannel.tryLock();
							if (fileLock != null)
								break;
							Thread.sleep(500L);
						}
						String s = null;
						while ((s = randomAccessFile.readLine()) != null) {
							String s1 = new String(s.getBytes("ISO-8859-1"), "utf-8");
							List<MessageDto> log2s = JSON.parseArray(s1, MessageDto.class);
							if (taskInfoQueue.size() > 2500) {
								Thread.sleep(3000L);
								continue;
							}
							taskInfoQueue.add(log2s);
						}
						fileLock.release();
						fileChannel.close();
						randomAccessFile.close();
						boolean bool = FileUtils.deleteQuietly(f);
					}
					Thread.sleep(3000L);
				} catch (Throwable t) {
					//log.error(" error : {} ", t.getMessage());
				}
			}

		}
	}

	public class TaskWriterRunner implements Runnable{
		private int name;
		public TaskWriterRunner( int name) {
			this.name = name;
		}
		@Override
		public void run() {
			//log.info(" {}", Integer.valueOf(this.name));
			while (true) {
				try {
					List<MessageDto> logs = taskInfoQueue.take();
					//log.info(" exit : {} {}{} ,{}", new Object[] { receive, Long.valueOf(exit.addAndGet(logs.size())), success, fileWriter });
					if (logs == null || logs.size() == 0) {
						Thread.sleep(1000L);
						continue;
					}
					sendOrWriter(logs);
				} catch (InterruptedException ex) {
					ex.printStackTrace();
					//log.error(" error ex : {} ", ex.getMessage());
				} catch (Throwable t) {
					t.printStackTrace();
					//log.error(" error t : {} ", t.getMessage());
				}
			}
		}
		private void sendOrWriter(List<MessageDto> logs) {
			int retryState = retry(logs);
			if (retryState != 0) {
				SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH-mm");
				String date = df.format(new Date());
				String fileName = fileDirector + File.separator + date + "_" + this.name + ".data";
				long start = System.currentTimeMillis();
				String json = JSON.toJSONString(logs);
				writerFile(json + "\n", fileName);
				//log.info(" fileWriter : {} {}{} ,{}", new Object[] { receive,exit, success, Long.valueOf(fileWriter.addAndGet(logs.size())) });
				long end = System.currentTimeMillis();
				//log.info(" {} ", Long.valueOf(end - start));
			}
		}

		private int retry(List<MessageDto> logs) {
			int state = 200;
			if (retryNum == 0)
				return state;
			long start = System.currentTimeMillis();
			for (int i = 0; i < retryNum; i++) {

				R r= null;
				try {
					r = remoteKafkaSendService.sendKafkas(logs, SecurityConstants.FROM_IN);
				} catch (Exception e) {
					break;
				}
				state=r.getCode();
				if (0==r.getCode()) {
					//log.info("success : {} {}{} ,{}", new Object[] { receive, exit, Long.valueOf(success.addAndGet(logs.size())), fileWriter });
					break;
				}
			}
			long end = System.currentTimeMillis();
			//log.info(" {}" , Long.valueOf(end - start));
			return state;
		}

	}
	public void submit(List<MessageDto> logs) {
		this.taskInfoQueue.add(logs);
		//log.info("receiver: {} {}{} ,{}", new Object[] { Long.valueOf(this.receive.addAndGet(logs.size())), this.exit, this.success, this.fileWriter });
	}

	public void writerFile(String content, String fileName) {
		FileChannel channel = null;
		FileLock lock = null;
		RandomAccessFile raf = null;
		try {
			raf = new RandomAccessFile(fileName, "rw");
			raf.seek(raf.length());
			channel = raf.getChannel();
			while (true) {
				lock = channel.tryLock();
				if (lock != null)
					break;
				Thread.sleep(500L);
			}
			ByteBuffer sendBuffer = ByteBuffer.wrap(content.getBytes("UTF-8"));
			channel.write(sendBuffer);
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			if (lock != null)
				try {
					lock.release();
				} catch (IOException e) {
					e.printStackTrace();
				}
			if (channel != null)
				try {
					channel.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			if (raf != null)
				try {
					raf.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
		}
	}
}
